

package com.hazelcast.jet.impl.processor;

import com.hazelcast.jet.core.AbstractProcessor;
import com.hazelcast.jet.core.ProcessorMetaSupplier;
import com.hazelcast.jet.core.ProcessorSupplier;

import javax.annotation.Nonnull;

/**
 * A processor that fails if it receives any input. The error message
 * suggests a different edge configuration should have been used so that no
 * data is received.
 * <p>
 * It ignores any restored state.
 * <p>
 * The processor is designed to be used as a dummy instance where only some
 * members have a real instances of another type.
 *
 * @see ProcessorMetaSupplier#forceTotalParallelismOne(ProcessorSupplier)
 */
public class ExpectNothingP extends AbstractProcessor {

    @Override
    protected boolean tryProcess(int ordinal, @Nonnull Object item) {
        throw new IllegalStateException("This vertex has a total parallelism of one and as such only" + " expects input on a specific node. Edge configuration must be adjusted" + " to make sure that only the expected node receives any input." + " Unexpected input received from ordinal " + ordinal + ": " + item);
    }

    @Override
    protected void restoreFromSnapshot(@Nonnull Object key, @Nonnull Object value) {
        // State might be broadcast to all instances including colleagues of
        // another type - ignore it in the expect-nothing instances
    }

    @Override
    public boolean closeIsCooperative() {
        return true;
    }
}
